package org.peoplesearch.collector;

import com.perm.kate.api.KException;
import org.peoplesearch.vk.executor.VkAPiExecutor;
import org.peoplesearch.vk.executor.VkCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class Collector {

    final static Logger logger = LoggerFactory.getLogger(Collector.class);

    private final VkAPiExecutor vkAPiExecutor;
    private volatile boolean which = true;
    private Long seed;
    private String data_dir;
    final Set<Long> seen = Collections.synchronizedSet(new HashSet<Long>());

    public Collector(VkAPiExecutor vkAPiExecutor, Long seed, String data_dir) {
        this.vkAPiExecutor = vkAPiExecutor;
        this.seed = seed;
        this.data_dir = data_dir;
    }

    public static void main(String[] args) {
        ApplicationContext ctx = new ClassPathXmlApplicationContext(
                "executor.xml",
                "collector.xml");
        Collector collector = (Collector) ctx.getBean("collector");
        collector.collectIds();
    }

    private BufferedWriter getRezultIdStore(String fileName) {
        BufferedWriter store;
        try {
            store = new BufferedWriter(new FileWriter(new File(data_dir + fileName + "_" + which)));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return store;
    }

    private void storeId(List<Long> idStore, String fileName) {
        BufferedWriter writer = getRezultIdStore(fileName);
        try {
            synchronized (seen) {
                writer.write(idStore.size() + "\n");
                for (Long id : idStore) {
                    writer.write(id + "\n");
                }
            }
            writer.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void switchFiles() {
        if (which == true) {
            which = false;
        } else which = true;
    }

    public void collectIds() {

        logger.debug("Seed is " + seed);

        final Queue<Long> usersQueue = new ConcurrentLinkedQueue<Long>();

        final AtomicInteger commandsCount = new AtomicInteger();
        final List<Long> IdsToStore = new LinkedList<Long>();

        usersQueue.add(seed);

        final VkCommand collectorCommand = new VkCommand() {
            public void run() {
                int commandId = commandsCount.incrementAndGet();
                String commandPrefix = "Command number " + commandId + " ";
                Long current = usersQueue.poll();
                if (commandId % 10 == 0) logger.debug(commandPrefix + "In this command current id is " + current);
                try {
                    List<Long> currentUsers = api.getOnlineFriends(current);
                    logger.trace(commandPrefix + " got user id's. Id's : " + currentUsers.size());
                    for (Long currentUID : currentUsers) {

                        if (!seen.contains(currentUID)) {
                            synchronized (seen) {
                                if (!seen.contains(currentUID)) {
                                    seen.add(currentUID);
                                    usersQueue.add(currentUID);
                                    IdsToStore.add(currentUID);
                                }
                            }
                        }
                        logger.trace("Addded user " + currentUID + ". number of commands is " + usersQueue.size());
                    }
                } catch (Exception e) {
                    logger.error(e.toString());
                    if (!((e instanceof KException) && ((KException) e).error_code == 15)) {
                        e.printStackTrace();
                        System.exit(-1);
                    }
                }
            }
        };


        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    logger.debug("Started storing in the files");
                    storeId(IdsToStore, "all");
                    switchFiles();
                    logger.debug("Completed storing in the files");
                } catch (Exception e) {
                    e.printStackTrace();
                    System.exit(-1);
                }
            }
        }, 20, 300, TimeUnit.SECONDS);

        while (usersQueue.size() > 0) {
            while (usersQueue.size() > 0) {
                if (vkAPiExecutor.getTasksInQueue() > 100500) {
                    vkAPiExecutor.waitForCompletion();
                }
                vkAPiExecutor.addCommand(collectorCommand);
            }
            vkAPiExecutor.waitForCompletion();
        }
        vkAPiExecutor.terminate(100);
    }

}
